package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo21Partitions {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo21Partitions")

    conf.set("spark.default.parallelism", "3")

    val sc: SparkContext = new SparkContext(conf)

    // 1、第一个RDD的分区数由切片的数量决定，但也会受到minPartitions的影响
    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")
    println(s"stuRDD的分区数为：${stuRDD.getNumPartitions}")

    // 2、默认情况下，子RDD的分区数等于父RDD的分区数
    val clazzKVRDD: RDD[(String, Int)] = stuRDD.map(line => (line.split(",")(4), 1))
    println(s"clazzKVRDD的分区数为：${clazzKVRDD.getNumPartitions}")

    // 3、shuffle类算子可以手动指定分区数量
    val grpRDD: RDD[(String, Iterable[Int])] = clazzKVRDD.groupByKey(4)
    println(s"grpRDD的分区数为：${grpRDD.getNumPartitions}")

    // 4、如果shuffle类算子没有指定分区数，默认等于父RDD的分区数
    // 当然也可以由Spark的一个参数决定：spark.default.parallelism
    // shuffle类算子最后返回的RDD分区数决定因素：手动指定 > spark.default.parallelism > 父RDD的分区数
    val grpRDD01: RDD[(String, Iterable[Int])] = clazzKVRDD.groupByKey()
    println(s"grpRDD01的分区数为：${grpRDD01.getNumPartitions}")

    // 可以通过coalesce、repartition方法手动改变RDD的分区数
    val filterRDD: RDD[String] = stuRDD.filter(line => line.split(",")(4) == "文科一班")
    println(s"filterRDD的分区数为：${filterRDD.getNumPartitions}")

    val coaFilterRDD: RDD[String] = filterRDD.coalesce(1)
    println(s"coaFilterRDD的分区数为：${coaFilterRDD.getNumPartitions}")

    // coalesce只能用于减少分区数量
    // coalesce默认不使用shuffle改变分区的数量，所以只适用于减少分区数
    val coaFilterRDD01: RDD[String] = filterRDD.coalesce(4)
    println(s"coaFilterRDD01的分区数为：${coaFilterRDD01.getNumPartitions}")

    /**
     * repartition实际上就是调用了coalesce方法，只不过将shuffle开关置为true
     * 即使用shuffle的方式改变分区数
     * 使用repartition不仅可以增加分区数，也可以减少分区数
     */
    val repFilterRDD01: RDD[String] = filterRDD.repartition(4)
    println(s"repFilterRDD01的分区数为：${repFilterRDD01.getNumPartitions}")

    val repFilterRDD02: RDD[String] = filterRDD.repartition(1)
    println(s"repFilterRDD02的分区数为：${repFilterRDD02.getNumPartitions}")

    /**
     * 如何选择coalesce还是repartition
     * 增加分区：repartition
     * 减少分区：
     *    后续是否要做shuffle类的操作：
     *        需要：repartition
     *        不需要：coalesce
     */


  }

}
